package com.yl.flink.sink;

import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import com.yl.util.FlinkUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.Statement;

public class TDengineSink extends RichSinkFunction<MultiDataEntity> {
    private final String jdbcUrl;
    private transient Statement statement;
    private transient Connection connection;

    public TDengineSink(ParameterTool params) {
        this.jdbcUrl = params.get(Const.TDENGINE_URL);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = FlinkUtil.getTDengineConn(this.jdbcUrl);
        statement = connection.createStatement();
    }

    @Override
    public void invoke(MultiDataEntity multiDataEntity, Context context) throws Exception {
        try {
            String sql = String.format("insert into %s.%s using %s.%s tags('%s') values(%d,%d,%f)", jdbcUrl);
            statement.executeUpdate(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
        if (statement != null) {
            statement.close();
        }
    }
}
